#!/usr/bin/env python3 """Merge faster-whisper segment timing with correct script text.""" import json, os BASE = r'E:\集群文件夹\factory_os\short_video_real_data_pipeline\phase4b_90s_formal_sample\v22_kimi_claude_loop\full_story_douyin_video' SUB_DIR = os.path.join(BASE, '04_subtitles', 'v2') # Load faster-whisper timing with open(os.path.join(SUB_DIR, 'captions_words_v2.json'), 'r', encoding='utf-8') as f: fw_data = json.load(f) # Correct script text SCRIPT_SEGMENTS = [ '我让 Claude Code 自己做视频。录屏失败、AI 跑偏、从 69 爬到 92。这不是特效，这是真实工作流。', 'Windows 后台抓不到桌面画面。试了五种录屏方案，全是空文件。录屏路线，彻底失败。', '更糟的是，Claude Code 跑去研究亚马逊选品。ChatGPT 看不下去：放弃录屏，让它自己画界面。', '于是用 Python 画终端界面、工具调用卡片。四个 AI 代理并行跑，进度条在涨，代码在敲。每条命令都来自真实执行日志。', '每轮渲染完 Kimi 审核打分。v1 69 分，改字体 v2 72，加动画 v3 跳到 81。修字幕 v4 85，砍时长 v6 89。V9 冲到 92。', '但下一轮有个硬编码没改。画面四个代理，底部只显示三个。92 直接砸到 76。', '修了六项回到 88，但回不到 92 了。一个硬编码，一夜回到解放前。最终决定：回滚 V9，冻结。', 'V9 上线审核门户，HTTP 200 通过。11 轮迭代，1 次崩溃，从 69 到 92。AI 自己审核自己，自己修复自己，自己部署自己。', '录屏失败、跑偏、崩过、又爬回来。执行、审核、修复、回归、交付。这才是真正的 AI 工作流。', ] fw_segments = fw_data['segments'] total_dur = fw_data['total_duration'] print(f'FW segments: {len(fw_segments)}, duration: {total_dur}s') # Keywords KEYWORD_MAP = { '录屏失败': '#FF4444', '录屏': '#FF4444', 'AI 跑偏': '#FF8800', '跑偏': '#FF8800', '真实执行日志': '#00DDFF', '真实日志': '#00DDFF', 'Kimi': '#00AAFF', '69': '#FFD700', '72': '#88FF88', '81': '#44FF44', '85': '#44FFDD', '89': '#44DDFF', '88': '#FFDD44', '92': '#FFD700', '76': '#FF2222', 'HTTP 200': '#44FF44', '200': '#44FF44', '回滚': '#44FF44', '冻结': '#44FFDD', '稳定交付': '#44FF44', '自动修复': '#44DDFF', } def is_highlight_char(ch, full_text): for kw in KEYWORD_MAP: clean_kw = kw.replace(' ', '') if clean_kw in full_text: idx = full_text.find(clean_kw) if idx != -1: # Check if char position falls within keyword range pos = full_text.find(ch) if pos != -1: for kw2 in KEYWORD_MAP: ckw = kw2.replace(' ', '') if ckw in full_text: k_start = full_text.find(ckw) k_end = k_start + len(ckw) if k_start <= pos < k_end: return True return False def format_time(sec): h = int(sec // 3600) m = int((sec % 3600) // 60) s = sec % 60 return f'{h:02d}:{m:02d}:{s:06.3f}' def srt_time(sec): return format_time(sec).replace('.', ',') def ass_time(sec): return format_time(sec) # Map FW segment boundaries to 9 script segments # Use character ratio to distribute script_chars = [len(s.replace(' ', '')) for s in SCRIPT_SEGMENTS] total_chars = sum(script_chars) # Calculate segment boundaries seg_boundaries = [0.0] cumulative = 0 for i, chars in enumerate(script_chars): cumulative += chars seg_boundaries.append(total_dur * cumulative / total_chars) seg_boundaries[-1] = total_dur # ensure last boundary equals total duration print(f'Segment boundaries: {[f"{b:.2f}" for b in seg_boundaries]}') # Build word-level segments segments_out = [] all_words = [] for seg_idx, script_text in enumerate(SCRIPT_SEGMENTS): seg_start = seg_boundaries[seg_idx] seg_end = seg_boundaries[seg_idx + 1] clean_text = script_text chars = list(clean_text) num_chars = len(chars) char_dur = (seg_end - seg_start) / max(num_chars, 1) words = [] # Track character position in the text for ci, ch in enumerate(chars): if ch == ' ': continue # skip spaces in word-level w_start = seg_start + ci * char_dur w_end = seg_start + (ci + 1) * char_dur # Check highlight is_hl = False for kw in KEYWORD_MAP: ckw = kw.replace(' ', '') if ckw in clean_text: k_start = clean_text.find(ckw) k_end = k_start + len(ckw) if k_start <= ci < k_end: is_hl = True break wd = {'word': ch, 'start': round(w_start, 2), 'end': round(w_end, 2), 'is_highlight': is_hl} words.append(wd) all_words.append(wd) segments_out.append({ 'start': round(seg_start, 2), 'end': round(seg_end, 2), 'text': script_text, 'words': words }) # Save updated JSON captions_words = { 'total_duration': round(total_dur, 1), 'model': 'faster-whisper-base-seg-timing + correct-script-text', 'segments': segments_out, 'word_count': len(all_words), 'note': 'Timing derived from faster-whisper base model on CUDA RTX3090, text from verified script' } with open(os.path.join(SUB_DIR, 'captions_words_v2.json'), 'w', encoding='utf-8') as f: json.dump(captions_words, f, ensure_ascii=False, indent=2) print(f'captions_words_v2.json saved: {len(all_words)} words, {len(segments_out)} segments') # Generate SRT - group into subtitle displays (max 2 lines, 13 chars/line) def split_into_lines(words_list, max_chars=26): """Group words into subtitle lines.""" groups = [] current_group = {'words': [], 'text': ''} for w in words_list: test_text = current_group['text'] + w['word'] if len(test_text) > max_chars and current_group['text']: groups.append(current_group) current_group = {'words': [w], 'text': w['word']} else: current_group['words'].append(w) current_group['text'] += w['word'] if current_group['text']: groups.append(current_group) return groups def split_line(line_text): """Split a line into two if over 13 chars.""" if len(line_text) <= 13: return [line_text] # Try to split at punctuation or midpoint mid = len(line_text) // 2 for j in range(mid, 0, -1): if line_text[j] in ' ，。！？、': return [line_text[:j+1].strip(), line_text[j+1:].strip()] return [line_text[:mid], line_text[mid:]] # Build SRT srt_entries = [] sub_idx = 1 for seg in segments_out: groups = split_into_lines(seg['words'], 26) for group in groups: grp_start = group['words'][0]['start'] grp_end = group['words'][-1]['end'] if grp_end - grp_start < 0.5: grp_end = grp_start + 0.5 lines = split_line(group['text']) srt_entries.append(f'{sub_idx}') srt_entries.append(f'{srt_time(grp_start)} --> {srt_time(grp_end)}') srt_entries.extend(lines) srt_entries.append('') sub_idx += 1 srt_path = os.path.join(SUB_DIR, 'captions_final_v2.srt') with open(srt_path, 'w', encoding='utf-8') as f: f.write('\n'.join(srt_entries)) print(f'SRT saved: {sub_idx-1} entries -> {srt_path}') # Generate ASS with highlighting ass_header = '''[Script Info] ScriptType: v4.00+ PlayResX: 1080 PlayResY: 1920 ScaledBorderAndShadow: yes [V4+ Styles] Format: Name, Fontname, Fontsize, PrimaryColour, SecondaryColour, OutlineColour, BackColour, Bold, Italic, Underline, StrikeOut, ScaleX, ScaleY, Spacing, Angle, BorderStyle, Outline, Shadow, Alignment, MarginL, MarginR, MarginV, Encoding Style: Default,Microsoft YaHei,30,&H00FFFFFF,&H0000FF00,&H00000000,&H80000000,0,0,0,0,100,100,0,0,1,2,0,2,50,50,80,134 Style: Highlight,Microsoft YaHei,30,&H0000FFDD,&H0000FF00,&H00000000,&H80000000,1,0,0,0,100,100,0,0,1,2,0,2,50,50,80,134 [Events] Format: Layer, Start, End, Style, Name, MarginL, MarginR, MarginV, Effect, Text ''' def apply_ass_highlight(text): sorted_kws = sorted(KEYWORD_MAP.items(), key=lambda x: -len(x[0])) result = [] remaining = text while remaining: best_pos = len(remaining) best_kw = None best_color = None for kw, color in sorted_kws: pos = remaining.find(kw) if pos != -1 and pos < best_pos: best_pos = pos best_kw = kw best_color = color if best_kw is not None: if best_pos > 0: result.append(remaining[:best_pos]) bgr = best_color[5:7] + best_color[3:5] + best_color[1:3] result.append(fr"{{\c&H{bgr}&\b1}}{best_kw}{{\c&HFFFFFF&\b0}}") remaining = remaining[best_pos + len(best_kw):] else: result.append(remaining) break return ''.join(result) ass_lines = [ass_header] for seg in segments_out: groups = split_into_lines(seg['words'], 26) for group in groups: grp_start = group['words'][0]['start'] grp_end = group['words'][-1]['end'] if grp_end - grp_start < 0.5: grp_end = grp_start + 0.5 lines = split_line(group['text']) hl_lines = [apply_ass_highlight(l) for l in lines] ass_text = '\\N'.join(hl_lines) has_hl = any('\\\\c&H' in l for l in hl_lines) style_name = 'Highlight' if has_hl else 'Default' ass_lines.append(f'Dialogue: 0,{ass_time(grp_start)},{ass_time(grp_end)},{style_name},,0,0,0,,{ass_text}') ass_path = os.path.join(SUB_DIR, 'captions_final_v2.ass') with open(ass_path, 'w', encoding='utf-8') as f: f.write('\n'.join(ass_lines)) print(f'ASS saved: {ass_path}') print('V2 subtitle generation with FW timing complete!')